package com.yjd.bi.etl.canal;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.yjd.bi.canal.model.CanalMessageModel;
import com.yjd.bi.canal.service.ICanalMessageService;
import com.yjd.comm.annotation.ServiceRemoteProxy;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.job.BaseJob;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;

public class CanalDataTransferLogJob extends BaseJob{

	private static int GEN_BANCH=500;
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	ICanalMessageService canalMessageService;
	
	@Override
	protected boolean doExecute() throws Exception {
		transfer();
		return true;
	}

	@Override
	protected String getJobType() {
		return CanalDataTransferLogJob.class.getName();
	}

	
	/**
	 * 
	 * @return
	 * @throws Exception 
	 */
	public boolean transfer() throws Exception{
		Map paramMap=FrameUtil.newHashMap("status",1);
		Long sum=canalMessageService.getModelListCount(paramMap,DbWREnums.WRITE);
		List<CanalMessageModel> list=new ArrayList<CanalMessageModel>();
		if(sum==0){
			return true;
		}
		/**
		 * 保留最后一条
		 */
		while (sum!=1){
			logger.info("cannal数据转移中..."+sum);
			paramMap.put("dbtype","mysql");
			paramMap.put("rowstartindex",0);
			paramMap.put("rowendindex", GEN_BANCH>=sum?sum-1:GEN_BANCH);
			list=canalMessageService.getModelList(paramMap,DbWREnums.WRITE);
			canalMessageService.moveToLog(list);
			paramMap=FrameUtil.newHashMap("status",1);
			sum=canalMessageService.getModelListCount(paramMap,DbWREnums.WRITE);
		}
		return true;
	}
}
